package com.my;

import com.alibaba.fastjson.JSONObject;
import com.my.serialize.MySqlDeserialization;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestClientBuilder;

import java.util.ArrayList;

public class App {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //源算子
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
                .databaseList("db_device")
                .hostname("192.168.80.101")
                .port(3308)
                .tableList("tb_user")
                .username("root")
                .password("root")
                .deserializer(new StringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> streamSource = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource");
        //输出算子
        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("192.168.80.101",9200,"Http"));
        ElasticsearchSink.Builder<String> esBuilser = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<String>() {
                    @Override
                    public void process(String element, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                        JSONObject jsonObject = JSONObject.parseObject(element);
                        IndexRequest source = Requests.indexRequest().index("test_index").type("test_type").source(jsonObject);
                        requestIndexer.add(source);
                    }
                }
        );
        esBuilser.setRestClientFactory(new RestClientFactory() {
            @Override
            public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
                restClientBuilder.setHttpClientConfigCallback(
                        new RestClientBuilder.HttpClientConfigCallback() {
                            @Override
                            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                                BasicCredentialsProvider provider = new BasicCredentialsProvider();
                                provider.setCredentials(AuthScope.ANY,new UsernamePasswordCredentials("admin","123456"));
                                return httpAsyncClientBuilder.setDefaultCredentialsProvider(provider);
                            }
                        }
                );
            }
        });
        streamSource.addSink(esBuilser.build()).name("sink_es");
        env.execute("mysql_es");
    }
}
